/** 
* @version:1.0.1
* @Description: （对类进行功能描述）
* @author: yangdechao
* @date: datedate 2021年11月15日 上午11:00:30
*/
package cn.com.guage.flink.transformation;

import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

/**
 * @version:1.0.1
 * @Description: Union可以将两个或多个同数据类型的流合并成一个流。
 * @author: yangdechao
 * @date: datedate 2021年11月15日 上午11:00:30
 */
public class UnionTest {
	public static boolean runingFlag = true;

	public static void main(String[] args) throws Exception {
		StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
		// env.setParallelism(1);

		DataStreamSource<Tuple3<Integer, String, String>> ADataStream = env
				.addSource(new SourceFunction<Tuple3<Integer, String, String>>() {
					public void run(SourceContext<Tuple3<Integer, String, String>> ctx) throws Exception {
						while (runingFlag) {
							ctx.collect(new Tuple3<Integer, String, String>(1, "111", "A"));
							Thread.sleep(500);
						}
					}

					public void cancel() {
						runingFlag = false;
					}
				});
		DataStreamSource<Tuple3<Integer, String, String>> BDataStream = env
				.addSource(new SourceFunction<Tuple3<Integer, String, String>>() {
					public void run(SourceContext<Tuple3<Integer, String, String>> ctx) throws Exception {
						while (runingFlag) {
							ctx.collect(new Tuple3<Integer, String, String>(1, "222", "B"));
							Thread.sleep(500);
						}
					}

					public void cancel() {
						runingFlag = false;
					}
				});
		DataStreamSource<Tuple3<Integer, String, String>> CDataStream = env
				.addSource(new SourceFunction<Tuple3<Integer, String, String>>() {
					/**
					 * 
					 */
					private static final long serialVersionUID = 6527710539190800107L;

					public void run(SourceContext<Tuple3<Integer, String, String>> ctx) throws Exception {
						while (runingFlag) {
							ctx.collect(new Tuple3<Integer, String, String>(1, "333", "C"));
							Thread.sleep(500);
						}
					}

					public void cancel() {
						runingFlag = false;
					}
				});

		ADataStream.union(BDataStream).union(CDataStream).print();
		env.execute("测试合流");
	}
}
